{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "5dcc7471",
   "metadata": {},
   "source": [
    "# Unstructured Data Ingestion and Processing With Ray Data\n",
    "\n",
    "<div align=\"left\">\n",
    "<a target=\"_blank\" href=\"https://console.anyscale.com/template-preview/unstructured-data-ingestion\"><img src=\"https://img.shields.io/badge/🚀 Run_on-Anyscale-9hf\"></a>&nbsp;\n",
    "<a href=\"https://github.com/ray-project/ray/tree/master/doc/source/data/examples/unstructured-data-ingestion\" role=\"button\"><img src=\"https://img.shields.io/static/v1?label=&amp;message=View%20On%20GitHub&amp;color=586069&amp;logo=github&amp;labelColor=2f363d\"></a>&nbsp;\n",
    "</div>\n",
    "\n",
    "**Time to complete**: 35 min | **Difficulty**: Advanced | **Prerequisites**: Data engineering experience, document processing, basic natural language processing (NLP) knowledge\n",
    "\n",
    "Build a comprehensive document ingestion pipeline that transforms unstructured documents from data lakes into structured, analytics-ready datasets using Ray Data's distributed processing capabilities for enterprise data warehouse workflows."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5311c357",
   "metadata": {},
   "source": [
    "## Table of contents\n",
    "\n",
    "1. [Data lake document discovery](#step-1-data-lake-document-discovery) (8 min)\n",
    "2. [Document processing and classification](#step-2-document-processing-and-classification) (10 min)\n",
    "3. [Text extraction and enrichment](#step-3-text-chunking-and-enrichment) (8 min)\n",
    "4. [Data warehouse output](#step-4-data-warehouse-schema-and-output) (6 min)\n",
    "5. [Verification and summary](#verification) (3 min)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5431ee76",
   "metadata": {},
   "source": [
    "## Learning objectives\n",
    "\n",
    "**Why unstructured data ingestion matters**: Enterprise data lakes contain vast amounts of unstructured documents (PDFs, Word docs, presentations, reports) that need systematic processing to extract business value for analytics and reporting.\n",
    "\n",
    "**Ray Data's ingestion capabilities**: Distribute document processing across clusters to handle large-scale document collections, extract structured data, and prepare analytics-ready datasets for data warehouse consumption.\n",
    "\n",
    "**Data lake to warehouse patterns**: Techniques used by data engineering teams to systematically process document collections, extract structured information, and create queryable datasets for business intelligence.\n",
    "\n",
    "**Production ingestion workflows**: Scalable document processing patterns that handle diverse file formats, extract metadata, and create structured schemas for downstream analytics systems."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9f4dbb31",
   "metadata": {},
   "source": [
    "## Overview\n",
    "\n",
    "**Challenge**: Enterprise data lakes contain millions of unstructured documents (PDFs, Word docs, presentations) across multiple formats that need systematic processing to extract business value. Traditional document processing approaches struggle with:\n",
    "- **Scale**: Single-machine processing limits document volume\n",
    "- **Consistency**: Manual extraction creates inconsistent schemas\n",
    "- **Integration**: Complex infrastructure for analysis\n",
    "- **Warehouse integration**: Manual data modeling and ETL processes\n",
    "- **Increasing data sizes**: Unstructured data is growing rapidly, often overtaking structured data by 10x in sheer data volume.\n",
    "\n",
    "**Solution**: Ray Data enables end-to-end document ingestion pipelines with native distributed operations for processing millions of documents efficiently.\n",
    "\n",
    "- **Scale**: Using streaming execution and massive scalability capabilities to process petabytes or even exabytes of data.\n",
    "- **Consistency**: A flexible API for supporting quality checks through `map` as well as support for any type of PyArrow schema. All data read also has to be consistent or else the pipeline fails along with additional configuration options for this behavior.\n",
    "- **Integration**: Supports integration with all data types as well as any AI types, running on CPUs/GPUs/accelerators on cloud and on-prem\n",
    "- **Warehouse integration**: Pre-built connectors to popular data warehouses as well as having the ability to build custom connectors easily.\n",
    "- **Increasing data sizes**: Supports single node optimization to scaling across 10k+ nodes"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0a6d5da2",
   "metadata": {},
   "source": [
    "## Prerequisites checklist\n",
    "\n",
    "Before starting, ensure you have:\n",
    "- [ ] An understanding of data lake and data warehouse concepts\n",
    "- [ ] Experience with document processing and text extraction\n",
    "- [ ] Knowledge of structured data formats (Parquet, Delta Lake, Iceberg)\n",
    "- [ ] A Python environment with Ray Data and document processing libraries\n",
    "- [ ] Access to S3 or other cloud storage for document sources"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "394fda83",
   "metadata": {},
   "source": [
    "Setup and initialize Ray Data:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fa1d9886",
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "import json \n",
    "import uuid  \n",
    "from datetime import datetime  \n",
    "from pathlib import Path \n",
    "from typing import Any, Dict, List  #\n",
    "\n",
    "import numpy as np \n",
    "import pandas as pd \n",
    "\n",
    "import ray \n",
    "from ray.data.aggregate import Count, Max, Mean, Sum \n",
    "from ray.data.expressions import col, lit \n",
    "\n",
    "ctx = ray.data.DataContext.get_current()\n",
    "\n",
    "# Disable progress bars for cleaner output in production\n",
    "# You can enable these for debugging: set to True to see progress\n",
    "ctx.enable_progress_bars = False\n",
    "ctx.enable_operator_progress_bars = False\n",
    "\n",
    "# Initialize Ray cluster connection to set the Ray Data context\n",
    "# Use runtime env to install dependencies across all workers\n",
    "runtime_env = dict(\n",
    "    pip= {\n",
    "        \"packages\": [\"unstructured[all-docs]==0.18.21\", \"pandas==2.3.3\"],\n",
    "        \"pip_install_options\": [\"--force-reinstall\", \"--no-cache-dir\"]\n",
    "    }\n",
    ")\n",
    "ray.init(ignore_reinit_error=True, runtime_env= runtime_env)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7499c63a",
   "metadata": {},
   "source": [
    "## Step 1: Data Lake Document Discovery"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7e4fb4f2",
   "metadata": {},
   "source": [
    "### Discover document collections in data lake"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fcdb1f64",
   "metadata": {
    "lines_to_next_cell": 1
   },
   "outputs": [],
   "source": [
    "# READ DOCUMENTS FROM DATA LAKE (S3)\n",
    "\n",
    "# Use Ray Data's read_binary_files() to load documents from S3\n",
    "# Why read_binary_files()?\n",
    "#   - Reads files as raw bytes (works for PDFs, DOCX, images, etc.)\n",
    "#   - Distributes file reading across cluster workers\n",
    "#   - include_paths=True gives us the file path for each document\n",
    "#\n",
    "# Parameters explained:\n",
    "#   - S3 path: Location of document collection in data lake\n",
    "#   - include_paths=True: Adds 'path' column with file location\n",
    "#   - ray_remote_args: Resource allocation per task\n",
    "#     * num_cpus=0.025: Very low CPU since this is I/O-bound (reading files)\n",
    "#     * This allows many concurrent reads without CPU bottleneck\n",
    "#   - limit(100): Process only 100 documents for this demo\n",
    "#     * Remove this limit to process entire collection\n",
    "\n",
    "document_collection = ray.data.read_binary_files(\n",
    "    \"s3://anyscale-rag-application/1000-docs/\",\n",
    "    include_paths=True,\n",
    "    ray_remote_args={\"num_cpus\": 0.025}\n",
    ").limit(100)\n",
    "\n",
    "# Display the schema to understand our data structure\n",
    "# At this point, we have: 'bytes' (file content) and 'path' (file location)\n",
    "print(f\"Dataset schema: {document_collection.schema()}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "49dac4b9",
   "metadata": {},
   "source": [
    "### Document metadata extraction"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b900fcf8",
   "metadata": {},
   "outputs": [],
   "source": [
    "# TEXT EXTRACTION FUNCTION\n",
    "# This function extracts text from various document formats (PDF, DOCX, etc.)\n",
    "# It processes one document at a time (distirbuted by Ray) and returns structured metadata + text\n",
    "\n",
    "\n",
    "def process_file(record: Dict[str, Any]) -> Dict[str, Any]:\n",
    "    \"\"\"\n",
    "    Extract text content from document files using the Unstructured library.\n",
    "    \n",
    "    BEGINNER NOTE:\n",
    "    - Input: A dictionary (record) with 'bytes' (file content) and 'path' (file location)\n",
    "    - Output: A dictionary with extracted text + metadata\n",
    "    - This function runs on each worker node in parallel\n",
    "    \n",
    "    Why extract text immediately?\n",
    "    - Avoids passing large binary data through multiple operations\n",
    "    - Reduces memory usage in downstream processing\n",
    "    - Enables faster processing by dropping binary data early\n",
    "    \"\"\"\n",
    "    # Import libraries inside function so each Ray worker has access\n",
    "    import io\n",
    "    from pathlib import Path\n",
    "    \n",
    "    from unstructured.partition.auto import partition\n",
    "    \n",
    "    # Extract file metadata from the input record\n",
    "    file_path = Path(record[\"path\"])  # Convert path string to Path object\n",
    "    file_bytes = record[\"bytes\"]  # Raw file content (binary data)\n",
    "    file_size = len(file_bytes)  # Size in bytes\n",
    "    file_extension = file_path.suffix.lower()  # Get extension (.pdf, .docx, etc.)\n",
    "    file_name = file_path.name  # Just the filename (not full path)\n",
    "    \n",
    "    # We can only extract text from certain file types\n",
    "    # If unsupported, return metadata with empty text\n",
    "    supported_extensions = {\".pdf\", \".docx\", \".doc\", \".pptx\", \".ppt\", \".html\", \".txt\"}\n",
    "    \n",
    "    if file_extension not in supported_extensions:\n",
    "        # Return a record with metadata but no extracted text\n",
    "        return {\n",
    "            \"document_id\": str(uuid.uuid4()),  # Generate unique ID\n",
    "            \"file_path\": str(file_path),\n",
    "            \"file_name\": file_name,\n",
    "            \"file_extension\": file_extension,\n",
    "            \"file_size_bytes\": file_size,\n",
    "            \"file_size_mb\": round(file_size / (1024 * 1024), 2),  # Convert to MB\n",
    "            \"discovery_timestamp\": datetime.now().isoformat(),\n",
    "            \"extracted_text\": \"\",  # Empty - unsupported format\n",
    "            \"text_length\": 0,\n",
    "            \"word_count\": 0,\n",
    "            \"extraction_status\": \"unsupported_format\"\n",
    "        }\n",
    "    \n",
    "    # Extract text using the Unstructured library\n",
    "    try:\n",
    "        # Create an in-memory file stream from bytes (avoids writing to disk)\n",
    "        with io.BytesIO(file_bytes) as stream:\n",
    "            # partition() automatically detects format and extracts text\n",
    "            # It returns a list of text elements (paragraphs, tables, etc.)\n",
    "            elements = partition(file=stream)\n",
    "            \n",
    "            # Combine all extracted text elements into one string\n",
    "            extracted_text = \" \".join([str(el) for el in elements]).strip()\n",
    "            \n",
    "            # Calculate text statistics for quality assessment\n",
    "            text_length = len(extracted_text)  # Total characters\n",
    "            word_count = len(extracted_text.split()) if extracted_text else 0\n",
    "            extraction_status = \"success\"\n",
    "            \n",
    "    except Exception as e:\n",
    "        # If extraction fails (corrupted file, unsupported format, etc.)\n",
    "        # Log the error and continue processing other files\n",
    "        print(f\"Cannot process file {file_path}: {e}\")\n",
    "        extracted_text = \"\"\n",
    "        text_length = 0\n",
    "        word_count = 0\n",
    "        extraction_status = f\"error: {str(e)[:100]}\"  # Store error message (truncated)\n",
    "    \n",
    "    # Return record with all metadata and extracted text\n",
    "    return {\n",
    "        \"document_id\": str(uuid.uuid4()),  # Unique identifier for this document\n",
    "        \"file_path\": str(file_path),\n",
    "        \"file_name\": file_name,\n",
    "        \"file_extension\": file_extension,\n",
    "        \"file_size_bytes\": file_size,\n",
    "        \"file_size_mb\": round(file_size / (1024 * 1024), 2),\n",
    "        \"discovery_timestamp\": datetime.now().isoformat(),\n",
    "        \"extracted_text\": extracted_text,  # The actual text content\n",
    "        \"text_length\": text_length,\n",
    "        \"word_count\": word_count,\n",
    "        \"extraction_status\": extraction_status\n",
    "    }\n",
    "\n",
    "\n",
    "# Ray Data's map() applies process_file() to each document in parallel\n",
    "# This is the \"embarrassingly parallel\" pattern - each document is independent\n",
    "print(\"Extracting text from documents...\")\n",
    "\n",
    "# Why map() instead of map_batches()?\n",
    "#   - map(): Process one record at a time (good for variable-size documents)\n",
    "#   - map_batches(): Process records in batches (better for vectorized operations)\n",
    "#   - Text extraction is I/O-bound and document-specific, so map() is ideal\n",
    "\n",
    "# Parameters:\n",
    "#   - process_file: The function to apply to each record\n",
    "\n",
    "documents_with_text = document_collection.map(\n",
    "    process_file\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "18f3c9e4",
   "metadata": {
    "lines_to_next_cell": 1
   },
   "outputs": [],
   "source": [
    "# Convert a sample of documents to pandas DataFrame for easy viewing\n",
    "documents_with_text.limit(25).to_pandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f7eb533a",
   "metadata": {},
   "outputs": [],
   "source": [
    "# BUSINESS METADATA ENRICHMENT FUNCTION (Modern Approach)\n",
    "# Instead of simple string matching, apply basic NLP for categorization.\n",
    "# For high-quality production, you'd use an ML or LLM model endpoint here.\n",
    "\n",
    "from transformers import pipeline\n",
    "\n",
    "# For demonstration: Use a small zero-shot classification model.\n",
    "# In a real pipeline, you should call a production LLM/ML endpoint or use a domain-specific model.\n",
    "# The 'facebook/bart-large-mnli' model works for zero-shot/label classification tasks.\n",
    "# You may swap with \"typeform/distilbert-base-uncased-mnli\" or another small MNLI model for lighter resource use.\n",
    "# If working at scale or with large docs, consider using Ray Serve + LLM API instead.\n",
    "\n",
    "zero_shot_classifier = pipeline(\n",
    "    \"zero-shot-classification\",\n",
    "    model=\"facebook/bart-large-mnli\"  # Or another MNLI model if needed\n",
    ")\n",
    "\n",
    "# Define candidate classes (map to business categories)\n",
    "CANDIDATE_LABELS = [\n",
    "    \"financial document\",  # maps to 'finance'\n",
    "    \"legal document\",      # maps to 'legal'\n",
    "    \"regulatory document\", # maps to 'compliance'\n",
    "    \"client document\",     # maps to 'client_services'\n",
    "    \"research document\",   # maps to 'research'\n",
    "    \"general document\"     # maps to 'general'\n",
    "]\n",
    "\n",
    "BUSINESS_CATEGORY_MAPPING = {\n",
    "    \"financial document\":    (\"financial_document\", \"finance\"),\n",
    "    \"legal document\":        (\"legal_document\", \"legal\"),\n",
    "    \"regulatory document\":   (\"regulatory_document\", \"compliance\"),\n",
    "    \"client document\":       (\"client_document\", \"client_services\"),\n",
    "    \"research document\":     (\"research_document\", \"research\"),\n",
    "    \"general document\":      (\"general_document\", \"general\"),\n",
    "}\n",
    "\n",
    "\n",
    "def enrich_business_metadata(record: Dict[str, Any]) -> Dict[str, Any]:\n",
    "    \"\"\"\n",
    "    Uses zero-shot text classification to predict business category and assign processing priority.\n",
    "    For production: Replace this with a call to your domain-tuned classifier or LLM endpoint.\n",
    "    \"\"\"\n",
    "\n",
    "    file_size = record[\"file_size_bytes\"]\n",
    "    text = record.get(\"extracted_text\", \"\") or \"\"\n",
    "    filename = record.get(\"file_name\", \"\")\n",
    "\n",
    "    # Concatenate extracted text with filename for context, up to limit (to save inference cost)\n",
    "    context_text = (filename + \"\\n\\n\" + text[:1000]).strip()  # Truncate to first 1000 chars for speed\n",
    "\n",
    "    # Run zero-shot classification (useful even with short context)\n",
    "    result = zero_shot_classifier(context_text, CANDIDATE_LABELS, multi_label=False)\n",
    "    top_label = result[\"labels\"][0] if result and \"labels\" in result and result[\"labels\"] else \"general document\"\n",
    "\n",
    "    doc_type, business_category = BUSINESS_CATEGORY_MAPPING.get(top_label, (\"general_document\", \"general\"))\n",
    "\n",
    "    # Priority assignment using simple logic + NLP keyword search (feel free to LLM this too)\n",
    "    lower_context = context_text.lower()\n",
    "    if any(w in lower_context for w in [\"urgent\", \"critical\", \"deadline\"]):\n",
    "        priority = \"high\"\n",
    "        priority_score = 3\n",
    "    elif any(w in lower_context for w in [\"important\", \"quarterly\", \"annual\"]):\n",
    "        priority = \"medium\"\n",
    "        priority_score = 2\n",
    "    else:\n",
    "        priority = \"low\"\n",
    "        priority_score = 1\n",
    "    \n",
    "    return {\n",
    "        **record,\n",
    "        \"document_type\": doc_type,\n",
    "        \"business_category\": business_category,\n",
    "        \"processing_priority\": priority,\n",
    "        \"priority_score\": priority_score,\n",
    "        \"estimated_pages\": max(1, file_size // 50000),\n",
    "        \"processing_status\": \"classified\"\n",
    "    }\n",
    "\n",
    "\n",
    "# Apply business metadata enrichment to all documents\n",
    "print(\"Enriching with business metadata (using zero-shot NLP)...\")\n",
    "\n",
    "# Note: zero-shot and LLM models can be heavy;\n",
    "# For fast local testing, use a smaller model, or replace with a production endpoint.\n",
    "\n",
    "documents_with_metadata = documents_with_text.map(\n",
    "    enrich_business_metadata\n",
    ")\n",
    "\n",
    "\n",
    "# View a few documents with business classification added\n",
    "documents_with_metadata.limit(5).to_pandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5d7464e5",
   "metadata": {
    "lines_to_next_cell": 1
   },
   "outputs": [],
   "source": [
    "\n",
    "# WHY AGGREGATIONS?\n",
    "# Before writing to warehouse, understand what we're processing:\n",
    "# - How many documents of each type?\n",
    "# - What's the size distribution?\n",
    "# - Which categories have the most content?\n",
    "\n",
    "# AGGREGATION 1: Document type distribution\n",
    "# Group by document_type and calculate statistics\n",
    "\n",
    "doc_type_stats = documents_with_metadata.groupby(\"document_type\").aggregate(\n",
    "    Count(),  # How many documents of each type?\n",
    "    Sum(\"file_size_bytes\"),  # Total size per document type\n",
    "    Mean(\"file_size_mb\"),  # Average size per document type\n",
    "    Max(\"estimated_pages\")  # Largest document per type\n",
    ")\n",
    "\n",
    "# AGGREGATION 2: Business category analysis\n",
    "# Understand the distribution across business categories\n",
    "# This helps with warehouse partitioning strategy\n",
    "\n",
    "category_stats = documents_with_metadata.groupby(\"business_category\").aggregate(\n",
    "    Count(),  # How many per category?\n",
    "    Mean(\"priority_score\"),  # Average priority per category\n",
    "    Sum(\"file_size_mb\")  # Total data volume per category\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0200de74",
   "metadata": {},
   "source": [
    "## Step 2: Document processing and classification\n",
    "\n",
    "### Text extraction and quality assessment"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "dc73f166",
   "metadata": {
    "lines_to_next_cell": 1
   },
   "outputs": [],
   "source": [
    "\n",
    "# QUALITY ASSESSMENT FUNCTION\n",
    "# Evaluate document quality to filter out low-quality or problematic documents\n",
    "\n",
    "def assess_document_quality(batch: pd.DataFrame) -> pd.DataFrame:\n",
    "    \"\"\"\n",
    "    Assess document quality for data warehouse ingestion.\n",
    "    \n",
    "    BEGINNER NOTE:\n",
    "    - Input: pandas DataFrame with multiple documents (a \"batch\")\n",
    "    - Output: Same DataFrame with quality assessment columns added\n",
    "    - Uses map_batches() for efficiency (process many docs at once)\n",
    "    \n",
    "    Why use map_batches() instead of map()?\n",
    "    - Batching is more efficient for lightweight operations\n",
    "    - Pandas DataFrame operations are optimized\n",
    "    - Reduces overhead from function calls\n",
    "    \n",
    "    Explicitly use batch_format=\"pandas\" \n",
    "    \"\"\"\n",
    "    quality_scores = np.zeros(len(batch), dtype=int)  # Numeric score (0-4)\n",
    "    quality_ratings = []  # Text rating (high/medium/low)\n",
    "    quality_issues_list = []  # List of issues found\n",
    "    \n",
    "    # Iterate through rows to apply business rules for quality\n",
    "    # Each document gets a score from 0-4 based on quality criteria\n",
    "    \n",
    "    for idx, row in batch.iterrows():\n",
    "        quality_score = 0\n",
    "        quality_issues = []\n",
    "        \n",
    "        # CRITERION 1: File size check\n",
    "        # Files smaller than 10KB might be empty or corrupt\n",
    "        if row[\"file_size_mb\"] > 0.01:  # More than 10KB\n",
    "            quality_score += 1\n",
    "        else:\n",
    "            quality_issues.append(\"file_too_small\")\n",
    "        \n",
    "        # CRITERION 2: Text length check\n",
    "        # Documents should have meaningful text content\n",
    "        if row[\"text_length\"] > 100:  # At least 100 characters\n",
    "            quality_score += 1\n",
    "        else:\n",
    "            quality_issues.append(\"insufficient_text\")\n",
    "        \n",
    "        # CRITERION 3: Business relevance check\n",
    "        # Classified documents are more valuable than unclassified\n",
    "        if row[\"business_category\"] != \"general\":\n",
    "            quality_score += 1\n",
    "        else:\n",
    "            quality_issues.append(\"low_business_relevance\")\n",
    "        \n",
    "        # CRITERION 4: Word count check\n",
    "        # Documents should have substantial content\n",
    "        if row[\"word_count\"] > 20:  # At least 20 words\n",
    "            quality_score += 1\n",
    "        else:\n",
    "            quality_issues.append(\"insufficient_content\")\n",
    "        \n",
    "        # Score 4: All checks passed - high quality\n",
    "        # Score 2-3: Some issues - medium quality\n",
    "        # Score 0-1: Major issues - low quality\n",
    "        quality_rating = \"high\" if quality_score >= 4 else \"medium\" if quality_score >= 2 else \"low\"\n",
    "        \n",
    "        # Store results for this document\n",
    "        quality_scores[idx] = quality_score\n",
    "        quality_ratings.append(quality_rating)\n",
    "        quality_issues_list.append(json.dumps(quality_issues))  # Convert list to JSON string\n",
    "    \n",
    "    batch[\"quality_score\"] = quality_scores\n",
    "    batch[\"quality_rating\"] = quality_ratings\n",
    "    batch[\"quality_issues\"] = quality_issues_list\n",
    "    \n",
    "    return batch\n",
    "\n",
    "\n",
    "# Apply quality assessment to all documents\n",
    "\n",
    "# Parameters:\n",
    "#   - batch_format=\"pandas\": Process as pandas DataFrame (easier than numpy arrays)\n",
    "#   - num_cpus=0.25: Very low CPU (this is lightweight logic)\n",
    "#   - batch_size=100: Process 100 documents at a time\n",
    "#     * Larger batches = fewer function calls = better efficiency\n",
    "\n",
    "quality_assessed_docs = documents_with_metadata.map_batches(\n",
    "    assess_document_quality,\n",
    "    batch_format=\"pandas\",  # Ray Data pattern: explicit pandas format\n",
    "    num_cpus=0.25,\n",
    "    batch_size=100\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6496dcaf",
   "metadata": {},
   "source": [
    "## Step 3: Text chunking and enrichment"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "81f39600",
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "# TEXT CHUNKING FUNCTION\n",
    "# Split long documents into smaller chunks for downstream processing\n",
    "# Why chunk? Many applications (LLMs, vector databases) have size limits\n",
    "\n",
    "def create_text_chunks(record: Dict[str, Any]) -> List[Dict[str, Any]]:\n",
    "    \"\"\"\n",
    "    Create overlapping text chunks from each document.\n",
    "    \n",
    "    BEGINNER NOTE:\n",
    "    - Input: ONE document record with full text\n",
    "    - Output: MULTIPLE chunk records (one document → many chunks)\n",
    "    - This is a \"one-to-many\" transformation\n",
    "    \n",
    "    Why chunking?\n",
    "    - LLM APIs have token limits (e.g., 4096 tokens)\n",
    "    - Vector databases work better with smaller chunks\n",
    "    - Enables more granular search and analysis\n",
    "    \n",
    "    Why overlapping chunks?\n",
    "    - Preserves context across chunk boundaries\n",
    "    - Prevents splitting important information\n",
    "    - 150 character overlap means each chunk shares text with neighbors\n",
    "    \"\"\"\n",
    "    text = record[\"extracted_text\"]  # The full document text\n",
    "    chunk_size = 1500  # Each chunk will be ~1500 characters\n",
    "    overlap = 150  # Adjacent chunks share 150 characters\n",
    "    \n",
    "    # Why these numbers?\n",
    "    # - 1500 chars ≈ 300-400 tokens (good for most LLM APIs)\n",
    "    # - 150 char overlap ≈ 10% overlap (preserves context without too much redundancy)\n",
    "    # - You can adjust these based on your use case\n",
    "    \n",
    "    # Create chunks of data by a sliding window\n",
    "    chunks = []\n",
    "    start = 0  # Starting position in the text\n",
    "    chunk_index = 0  # Track which chunk number this is\n",
    "\n",
    "    # There are many more advanced chunking methods, this example uses a simple technique for demo purposes\n",
    "    \n",
    "    # Loop until processing all the text\n",
    "    while start < len(text):\n",
    "        # Calculate end position (don't go past text length)\n",
    "        end = min(start + chunk_size, len(text))\n",
    "        \n",
    "        # Extract this chunk's text\n",
    "        chunk_text = text[start:end]\n",
    "        \n",
    "        # Create a new record for this chunk\n",
    "        # It contains all the original document metadata PLUS chunk-specific data\n",
    "        chunk_record = {\n",
    "            **record,  # All original fields (document_id, business_category, etc.)\n",
    "            \"chunk_id\": str(uuid.uuid4()),  # Unique ID for this specific chunk\n",
    "            \"chunk_index\": chunk_index,  # Position in sequence (0, 1, 2, ...)\n",
    "            \"chunk_text\": chunk_text,  # The actual text content of this chunk\n",
    "            \"chunk_length\": len(chunk_text),  # Characters in this chunk\n",
    "            \"chunk_word_count\": len(chunk_text.split())  # Words in this chunk\n",
    "        }\n",
    "        \n",
    "        chunks.append(chunk_record)\n",
    "        \n",
    "        # If you've reached the end of the text, you're done\n",
    "        if end >= len(text):\n",
    "            break\n",
    "        \n",
    "        # Move to next chunk position (with overlap)\n",
    "        # Example: If chunk_size=1500 and overlap=150\n",
    "        # Chunk 1: chars 0-1500\n",
    "        # Chunk 2: chars 1350-2850 (starts 150 before chunk 1 ended)\n",
    "        # Chunk 3: chars 2700-4200 (starts 150 before chunk 2 ended)\n",
    "        start = end - overlap\n",
    "        chunk_index += 1\n",
    "\n",
    "    # After creating all chunks, add how many chunks the document has\n",
    "    # This helps with progress tracking and completeness checks\n",
    "    for chunk in chunks:\n",
    "        chunk[\"total_chunks\"] = len(chunks)\n",
    "    \n",
    "    # Return the list of chunk records\n",
    "    # Ray Data's flat_map() automatically flattens this list\n",
    "    return chunks\n",
    "\n",
    "\n",
    "# Apply text chunking to all documents\n",
    "# Use flat_map() for one-to-many transformations\n",
    "# One document becomes multiple chunks\n",
    "print(\"Creating text chunks...\")\n",
    "\n",
    "# Why flat_map() instead of map()?\n",
    "#   - map(): One input → One output (document → document)\n",
    "#   - flat_map(): One input → Many outputs (document → multiple chunks)\n",
    "#   - flat_map() automatically \"flattens\" the list of chunks\n",
    "#\n",
    "# Example:\n",
    "#   Input: 100 documents\n",
    "#   Output: 10,000+ chunks (each document becomes ~100 chunks on average)\n",
    "#\n",
    "# Parameters:\n",
    "#   - num_cpus=0.5: Moderate CPU usage (string slicing is lightweight)\n",
    "\n",
    "chunked_documents = quality_assessed_docs.flat_map(\n",
    "    create_text_chunks,\n",
    "    num_cpus=0.5\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "71c548a1",
   "metadata": {},
   "source": [
    "## Step 4: Data warehouse schema and output\n",
    "\n",
    "### Create data warehouse schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3801f2fd",
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "# DATA WAREHOUSE SCHEMA TRANSFORMATION\n",
    "# Transform the raw processing data into a clean warehouse schema\n",
    "# This is the \"ETL\" part - Extract (done), Transform (now), Load (next)\n",
    "\n",
    "print(\"Creating data warehouse schema...\")\n",
    "\n",
    "\n",
    "# Get today's date in ISO format (YYYY-MM-DD)\n",
    "# Ray Data uses this to partition the data by date in the warehouse\n",
    "processing_date = datetime.now().isoformat()[:10]\n",
    "\n",
    "\n",
    "# Data warehouses need clean, organized schemas\n",
    "# Select only the columns needed and organize them logically\n",
    "#\n",
    "# Why not keep all columns?\n",
    "# - Cleaner schema = easier queries\n",
    "# - Less storage space\n",
    "# - Better performance\n",
    "# - Clear data contracts for downstream users\n",
    "\n",
    "warehouse_dataset = chunked_documents.select_columns([\n",
    "    # PRIMARY IDENTIFIERS: Keys for joining and relationships\n",
    "    \"document_id\",  # Links all chunks from same document\n",
    "    \"chunk_id\",     # Unique identifier for this specific chunk\n",
    "    \n",
    "    # DIMENSIONAL ATTRIBUTES: Categorical data for filtering/grouping\n",
    "    # These are typically used in WHERE clauses and GROUP BY\n",
    "    \"business_category\",      # finance, legal, compliance, etc.\n",
    "    \"document_type\",          # financial_document, legal_document, etc.\n",
    "    \"file_extension\",         # .pdf, .docx, etc.\n",
    "    \"quality_rating\",         # high, medium, low\n",
    "    \"processing_priority\",    # high, medium, low\n",
    "    \n",
    "    # FACT MEASURES: Numeric values for aggregation and analysis\n",
    "    # These are typically used in SUM(), AVG(), COUNT(), etc.\n",
    "    \"file_size_mb\",           # Document size\n",
    "    \"word_count\",             # Total words in document\n",
    "    \"chunk_word_count\",       # Words in this chunk\n",
    "    \"quality_score\",          # Numeric quality (0-4)\n",
    "    \"priority_score\",         # Numeric priority (1-3)\n",
    "    \"estimated_pages\",        # Page count estimate\n",
    "    \"chunk_index\",            # Position in document (0, 1, 2, ...)\n",
    "    \"total_chunks\",           # How many chunks total\n",
    "    \n",
    "    # CONTENT FIELDS: The actual data payload\n",
    "    \"chunk_text\",             # The text content (will rename this)\n",
    "    \"file_name\",              # Original filename\n",
    "    \"file_path\",              # S3 location\n",
    "    \n",
    "    # METADATA: Processing provenance and status tracking\n",
    "    \"discovery_timestamp\",    # When was file discovered\n",
    "    \"extraction_status\",      # success, error, unsupported_format\n",
    "    \"processing_status\"       # classified, processed, etc.\n",
    "])\n",
    "\n",
    "# RENAME COLUMNS for data warehouse conventions\n",
    "# \"chunk_text\" → \"text_content\" (more descriptive)\n",
    "warehouse_dataset = warehouse_dataset.rename_columns({\n",
    "    \"chunk_text\": \"text_content\"\n",
    "})\n",
    "\n",
    "# ADD PIPELINE METADATA: Constant columns for all records\n",
    "# These columns are the same for every record in this run\n",
    "# They help with data lineage and debugging\n",
    "warehouse_dataset = (\n",
    "    warehouse_dataset\n",
    "    .with_column(\"processing_date\", lit(processing_date))       # When was this processed?\n",
    "    .with_column(\"pipeline_version\", lit(\"1.0\"))               # Which version of pipeline?\n",
    "    .with_column(\"processing_engine\", lit(\"ray_data\"))         # What tool processed it?\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "181f85fb",
   "metadata": {},
   "source": [
    "### Write to data warehouse with partitioning"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1bd36e25",
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "# WRITE TO DATA WAREHOUSE - MAIN TABLE\n",
    "# Save all processed chunks to Parquet format with partitioning\n",
    "# This is the \"Load\" part of ETL\n",
    "\n",
    "# /mnt/cluster_storage is a shared storage volume accessible by all workers\n",
    "# In production, this would typically be:\n",
    "# - S3: s3://your-bucket/warehouse/\n",
    "# - Azure: abfs://container@account.dfs.core.windows.net/\n",
    "# - GCS: gs://your-bucket/warehouse/\n",
    "OUTPUT_WAREHOUSE_PATH = \"/mnt/cluster_storage\"\n",
    "\n",
    "# WRITE MAIN TABLE with PARTITIONING\n",
    "# write_parquet() is Ray Data's native way to save data\n",
    "\n",
    "# Key parameters explained:\n",
    "\n",
    "# partition_cols=[\"business_category\", \"processing_date\"]\n",
    "#   - Creates folder structure: business_category=finance/processing_date=2025-10-15/\n",
    "#   - Enables efficient querying: \"SELECT * FROM table WHERE business_category='finance'\"\n",
    "#   - Query engines (Spark, Presto, Athena) can skip entire partitions\n",
    "#   - Example structure:\n",
    "#       main_table/\n",
    "#       ├── business_category=finance/\n",
    "#       │   └── processing_date=2025-10-15/\n",
    "#       │       ├── part-001.parquet\n",
    "#       │       └── part-002.parquet\n",
    "#       ├── business_category=legal/\n",
    "#       │   └── processing_date=2025-10-15/\n",
    "#       │       └── part-001.parquet\n",
    "#       └── business_category=compliance/\n",
    "#           └── processing_date=2025-10-15/\n",
    "#               └── part-001.parquet\n",
    "\n",
    "# compression=\"snappy\"\n",
    "#   - Compress files to save storage space (50-70% reduction)\n",
    "#   - Snappy is fast and well-supported by all query engines\n",
    "#   - Alternatives: gzip (higher compression), zstd (good balance)\n",
    "\n",
    "# ray_remote_args={\"num_cpus\": 0.1}\n",
    "#   - Writing to storage is I/O-bound, not CPU-bound\n",
    "#   - Low CPU allocation allows more parallel writes\n",
    "\n",
    "warehouse_dataset.write_parquet(\n",
    "    f\"{OUTPUT_WAREHOUSE_PATH}/main_table/\",\n",
    "    partition_cols=[\"business_category\", \"processing_date\"],\n",
    "    compression=\"snappy\",\n",
    "    ray_remote_args={\"num_cpus\": 0.1}\n",
    ")\n",
    "\n",
    "print(\"Main warehouse table written successfully\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "54ca8014",
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "\n",
    "# CREATE BUSINESS-SPECIFIC DATASETS\n",
    "# Create specialized datasets for specific business teams\n",
    "# Each team gets only the data they need with relevant columns\n",
    "\n",
    "\n",
    "# Example: Compliance analytics dataset\n",
    "# Compliance team needs: document content, quality, and priority\n",
    "# Priority matters for compliance review workflows\n",
    "\n",
    "compliance_analytics = warehouse_dataset.filter(\n",
    "    lambda row: row[\"business_category\"] == \"compliance\"\n",
    ").select_columns([\n",
    "    \"document_id\",          # Link to main table\n",
    "    \"chunk_id\",             # Unique chunk identifier\n",
    "    \"text_content\",         # The actual text\n",
    "    \"quality_score\",        # Data reliability\n",
    "    \"processing_priority\",  # urgent/important/normal\n",
    "    \"processing_date\"       # When processed\n",
    "])\n",
    "\n",
    "# Write to dedicated compliance folder\n",
    "compliance_analytics.write_parquet(\n",
    "    f\"{OUTPUT_WAREHOUSE_PATH}/analytics/compliance/\",\n",
    "    partition_cols=[\"processing_date\"],\n",
    "    compression=\"snappy\",\n",
    "    ray_remote_args={\"num_cpus\": 0.1}\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cd8e1962",
   "metadata": {},
   "source": [
    "### Create analytics summary tables"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "add49b2a",
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "# CREATE ANALYTICS SUMMARY TABLES\n",
    "# Pre-compute common aggregations for fast dashboard queries\n",
    "# Summary tables = faster analytics queries\n",
    "\n",
    "\n",
    "# SUMMARY TABLE 1: Processing metrics by category and date\n",
    "\n",
    "# Answer questions like:\n",
    "# - How many documents processed per category per day?\n",
    "# - What's the total data volume per category?\n",
    "# - What's the average document quality by category?\n",
    "\n",
    "# This summary makes dashboard queries instant instead of scanning all data\n",
    "\n",
    "# groupby() groups data by multiple columns\n",
    "# aggregate() calculates statistics for each group\n",
    "\n",
    "processing_metrics = warehouse_dataset.groupby([\"business_category\", \"processing_date\"]).aggregate(\n",
    "    Count(),                    # How many chunks per category+date?\n",
    "    Sum(\"file_size_mb\"),        # Total data volume\n",
    "    Mean(\"word_count\"),         # Average document size\n",
    "    Mean(\"quality_score\")       # Average quality\n",
    ")\n",
    "\n",
    "# Write summary table\n",
    "# Smaller than main table, so queries are very fast\n",
    "processing_metrics.write_parquet(\n",
    "    f\"{OUTPUT_WAREHOUSE_PATH}/summaries/processing_metrics/\",\n",
    "    partition_cols=[\"processing_date\"],\n",
    "    compression=\"snappy\",\n",
    "    ray_remote_args={\"num_cpus\": 0.1}\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0e64e8b3",
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "# SUMMARY TABLE 2: Quality distribution\n",
    "# Answer questions like:\n",
    "# - What percentage of documents are high/medium/low quality?\n",
    "# - Which categories have the highest quality scores?\n",
    "# - How does quality correlate with document size?\n",
    "\n",
    "# This helps identify data quality issues by category\n",
    "\n",
    "quality_distribution = warehouse_dataset.groupby([\"quality_rating\", \"business_category\"]).aggregate(\n",
    "    Count(),                        # How many per quality+category?\n",
    "    Mean(\"word_count\"),             # Average document size by quality\n",
    "    Mean(\"chunk_word_count\")        # Average chunk size by quality\n",
    ")\n",
    "\n",
    "# Write quality summary\n",
    "# Used for quality monitoring dashboards\n",
    "\n",
    "quality_distribution.write_parquet(\n",
    "    f\"{OUTPUT_WAREHOUSE_PATH}/summaries/quality_distribution/\",\n",
    "    compression=\"snappy\",\n",
    "    ray_remote_args={\"num_cpus\": 0.1}\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6de9c8b7",
   "metadata": {},
   "source": [
    "## Verification\n",
    "\n",
    "After writing data to the warehouse, verify everything worked correctly. This section demonstrates:\n",
    "\n",
    "**Why verification matters:**\n",
    "- Ensures the pipeline wrote data successfully\n",
    "- Validates record counts match expectations\n",
    "- Confirms schema is correct\n",
    "- Provides sample data for visual inspection\n",
    "\n",
    "**What to verify:**\n",
    "1. Main table record count (should be 10,000+ chunks)\n",
    "2. Summary tables exist and have data\n",
    "3. Schema includes all expected columns\n",
    "4. Sample records look correct\n",
    "\n",
    "### Verify data warehouse outputs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8c58790c",
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "# VERIFY DATA WAREHOUSE OUTPUT\n",
    "# Always verify your data pipeline worked correctly\n",
    "# This is a critical production practice\n",
    "print(\"Verifying data warehouse integration...\")\n",
    "\n",
    "# Use Ray Data's read_parquet() to read what was just written\n",
    "# This verifies:\n",
    "# 1. Files were written successfully\n",
    "# 2. Partitioning works correctly\n",
    "# 3. Data can be read back (no corruption)\n",
    "#\n",
    "# Ray Data automatically discovers all partitions:\n",
    "# - main_table/business_category=finance/processing_date=2025-10-15/*.parquet\n",
    "# - main_table/business_category=legal/processing_date=2025-10-15/*.parquet\n",
    "# - etc.\n",
    "main_table_verify = ray.data.read_parquet(\n",
    "    f\"{OUTPUT_WAREHOUSE_PATH}/main_table/\",\n",
    "    ray_remote_args={\"num_cpus\": 0.025}  # Low CPU for reading\n",
    ")\n",
    "\n",
    "# Verify our aggregated metrics tables also wrote successfully\n",
    "metrics_verify = ray.data.read_parquet(\n",
    "    f\"{OUTPUT_WAREHOUSE_PATH}/summaries/processing_metrics/\",\n",
    "    ray_remote_args={\"num_cpus\": 0.025}\n",
    ")\n",
    "\n",
    "print(f\"Data warehouse verification:\")\n",
    "print(f\"  Main table records: {main_table_verify.count():,}\")\n",
    "print(f\"  Processing metrics: {metrics_verify.count():,}\")\n",
    "print(f\"  Schema compatibility: Verified\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ab152968",
   "metadata": {},
   "outputs": [],
   "source": [
    "# INSPECT SAMPLE DATA\n",
    "\n",
    "# take(10) gets first 10 records for manual inspection\n",
    "# This helps catch issues like:\n",
    "# - Wrong data types\n",
    "# - Missing fields\n",
    "# - Incorrect values\n",
    "# - Encoding problems\n",
    "samples = main_table_verify.take(10)\n",
    "\n",
    "# Display key fields from each sample record\n",
    "for i, record in enumerate(samples):\n",
    "    # Show abbreviated document ID (first 8 characters)\n",
    "    doc_id = record['document_id'][:8]\n",
    "    category = record['business_category']\n",
    "    words = record['word_count']\n",
    "    quality = record['quality_rating']\n",
    "    \n",
    "    print(f\"\\t{i+1}. Doc: {doc_id}, Category: {category}, Words: {words}, Quality: {quality}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "81f2d389",
   "metadata": {},
   "source": [
    "## Summary\n",
    "\n",
    "You built a complete end-to-end document ingestion pipeline using Ray Data. This section reviews what you learned and where to go from here.\n",
    "\n",
    "### What you built\n",
    "\n",
    "**Complete ETL pipeline**: Extract → Transform → Load\n",
    "1. **Extract**: Read 100 documents from S3 data lake\n",
    "2. **Transform**: Extract text, classify, assess quality, create chunks\n",
    "3. **Load**: Write to partitioned data warehouse with analytics tables\n",
    "\n",
    "**Final output**: From raw documents to structured warehouse\n",
    "- **Main table**: 10,000+ text chunks ready for analysis\n",
    "- **Business datasets**: Finance and compliance specific views\n",
    "- **Summary tables**: Pre-computed metrics for dashboards\n",
    "- **Partitioned storage**: Optimized for query performance\n",
    "\n",
    "### Ray Data operations you used\n",
    "\n",
    "This pipeline demonstrated all major Ray Data operations:\n",
    "\n",
    "| Operation | Purpose | When to use |\n",
    "|-----------|---------|-------------|\n",
    "| `read_binary_files()` | Load documents from S3/storage | Reading PDFs, images, any binary files |\n",
    "| `map()` | Transform each record individually | Variable-size processing, I/O-bound tasks |\n",
    "| `map_batches()` | Transform records in batches | Batch-optimized operations, ML inference |\n",
    "| `flat_map()` | One-to-many transformations | Chunking, splitting, exploding data |\n",
    "| `filter()` | Keep/remove records | Selecting subsets, data quality filtering |\n",
    "| `select_columns()` | Choose specific fields | Schema projection, reducing data size |\n",
    "| `rename_columns()` | Change column names | Schema standardization |\n",
    "| `groupby().aggregate()` | Calculate statistics | Analytics, metrics, summaries |\n",
    "| `write_parquet()` | Save to warehouse | Final output, checkpointing |\n",
    "\n",
    "### Key concepts for beginners\n",
    "\n",
    "**1. Distributed processing**\n",
    "- Your code runs on a cluster of machines (not just one)\n",
    "- Ray Data automatically distributes work across workers\n",
    "- Each function (process_file, assess_quality) runs in parallel\n",
    "- 100 documents processed simultaneously = 100x faster\n",
    "\n",
    "**2. Lazy evaluation**\n",
    "- Operations like `map()` and `filter()` don't execute immediately\n",
    "- Ray builds a plan and optimizes it\n",
    "- Execution happens when you call `write_parquet()`, `count()`, or `take()`\n",
    "- This allows Ray to optimize the entire pipeline\n",
    "\n",
    "**3. Resource management**\n",
    "- `batch_size`: How many records per batch\n",
    "- `concurrency`: How many tasks run in parallel (advanced)\n",
    "- `num_cpus`: How many CPU cores per task (advanced)\n",
    "- Balance these based on your workload\n",
    "\n",
    "**4. Partitioning strategy**\n",
    "- Partitions = folders organized by column values\n",
    "- `partition_cols=[\"business_category\", \"processing_date\"]`\n",
    "- Query engines skip entire folders when filtering\n",
    "- Enables efficient query performance by reducing data scanned\n",
    "\n",
    "### Implementation patterns applied\n",
    "\n",
    "**Code organization**:\n",
    "- Separate functions for each processing stage\n",
    "- Clear docstrings explaining purpose\n",
    "- Type hints for inputs and outputs\n",
    "- Comments explaining \"why\" not just \"what\"\n",
    "\n",
    "**Ray Data implementation patterns**:\n",
    "- Use `batch_format=\"pandas\"` for clarity\n",
    "- Process text early (don't pass binary through pipeline)\n",
    "- Appropriate resource allocation per operation type\n",
    "- Partition writes for query optimization\n",
    "- Use native Ray Data operations (not custom code)\n",
    "\n",
    "**Data engineering patterns**:\n",
    "- Immediate text extraction (reduces memory)\n",
    "- Separate classification stage (easier debugging)\n",
    "- Quality assessment (data validation)\n",
    "- Schema transformation (clean warehouse schema)\n",
    "- Verification step (always check output)\n",
    "\n",
    "### Production recommendations\n",
    "\n",
    "**Scaling to production:**\n",
    "\n",
    "1. **Remove the `.limit(100)` to process full dataset**\n",
    "   - Currently processing 100 docs for demo\n",
    "   - Remove this to process millions of documents\n",
    "   - No code changes needed, just remove one line\n",
    "\n",
    "2. **Tune resource parameters for your cluster**\n",
    "   ```python\n",
    "   # For larger clusters, increase parallelism:\n",
    "   batch_size=5000    # Larger batches\n",
    "   concurrency=50     # More parallel tasks (advanced)\n",
    "   num_cpus=2         # More CPU per task (advanced)\n",
    "   ```\n",
    "\n",
    "3. **Add error handling and retry logic**\n",
    "   ```python\n",
    "   # For production, catch specific errors:\n",
    "   try:\n",
    "       elements = partition(file=stream)\n",
    "   except CorruptedFileError:\n",
    "       # Log and skip\n",
    "   except TimeoutError:\n",
    "       # Retry with backoff\n",
    "   ```\n",
    "\n",
    "4. **Monitor with Ray Dashboard**\n",
    "   - View real-time progress\n",
    "   - Check resource utilization\n",
    "   - Identify bottlenecks\n",
    "   - Debug failures\n",
    "\n",
    "5. **Implement incremental processing**\n",
    "   ```python\n",
    "   # Only process new documents:\n",
    "   new_docs = all_docs.filter(\n",
    "       lambda row: row[\"processing_date\"] > last_run_date\n",
    "   )\n",
    "   ```\n",
    "\n",
    "6. **Add data quality checks**\n",
    "   - Validate schema before writing\n",
    "   - Check for null values\n",
    "   - Verify foreign key relationships\n",
    "   - Monitor quality metrics over time\n",
    "\n",
    "### What you learned\n",
    "\n",
    "**Ray Data fundamentals**:\n",
    "- How to read from cloud storage (S3)\n",
    "- Distributed data processing patterns\n",
    "- Batch versus row-based operations\n",
    "- Resource management and tuning\n",
    "- Writing to data warehouses\n",
    "\n",
    "**Data engineering skills**:\n",
    "- ETL pipeline design\n",
    "- Document processing at scale\n",
    "- Quality assessment strategies\n",
    "- Data warehouse schema design\n",
    "- Partitioning for performance\n",
    "\n",
    "**Production practices**:\n",
    "- Verification and testing\n",
    "- Error handling approaches\n",
    "- Resource optimization\n",
    "- Monitoring and debugging\n",
    "- Scalability considerations\n"
   ]
  }
 ],
 "metadata": {
  "jupytext": {
   "cell_metadata_filter": "-all",
   "main_language": "python",
   "notebook_metadata_filter": "-all"
  },
  "kernelspec": {
   "display_name": "base",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.12.11"
  },
  "orphan": true
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
